package com.doit.sparksql.day03

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{DataFrame, Encoder, Encoders, SparkSession}
p

/**
 * @DATE 2022/1/16/9:27
 * @Author MDK
 * @Version 2021.2.2
 *
 *    UDF
 *    UDAF  聚合函数
 *      spark内部自己的聚合函数 min max avg sum collect_list
 * */
object Demo02 {
  Logger.getLogger("org").setLevel(Level.ERROR)
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession.builder()
      .master("local")
      .appName("相似度")
      .getOrCreate()
    val df: DataFrame = spark.read.option("header", true).option("inferSchema",true).csv("sql_data/csv/b.csv")
    import org.apache.spark.sql.functions._
    import spark.implicits._
    //注册函数
    spark.udf.register("my_avg",udaf(MyAvgFunction))

    //调用聚合函数
    df.createTempView("user")
    spark.sql(
      """
        |select
        |gender,
        |my_avg(age)
        |from user
        |group by gender
        |""".stripMargin).show()
    spark.close()
  }
}
/**
 *   UDAF多个输入  一个输出结果  称聚合函数
 * Aggregator[IN, BUFFER, OUT]
 * 求平均年龄
 * 参数一  输入的数据类型  聚合函数的参数数据类型
 * 参数二  中间计算的缓存数据类型   (参数三  次数)
 * 参数四  输出结果类型
 * */
object MyAvgFunction extends Aggregator[Int, (Int, Int), Double] {
  override def zero: (Int, Int) = (0,0)

  //局部计算逻辑
  override def reduce(b: (Int, Int), a: Int): (Int, Int) = {
    (b._1+a, b._2+1)
  }

  //全局计算逻辑板
  override def merge(b1: (Int, Int), b2: (Int, Int)): (Int, Int) = {
    (b1._1+b2._1, b1._2+b2._2)
  }

  //最终的输出结果
  override def finish(reduction: (Int, Int)): Double = {
    reduction._1/reduction._2
  }

  //缓存的数据结构
  override def bufferEncoder: Encoder[(Int, Int)] = Encoders.tuple(Encoders.scalaInt, Encoders.scalaInt)

  //输出的数据结构
  override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}
